
#include "port.h"
#include "luas.h"
#include "cbuf.h"
#include "dict.h"
#include "pkt.h"

#define MAX_PORT 64
#define MAX_LINE 1024

static Port s_ports[MAX_PORT];
static Line s_lines[MAX_LINE];

#define fd2port(fd)  (&s_ports[fd])
#define fd2line(fd)  (&s_lines[fd])



static dict *s_name_port;



static int line_new()
{
    int fd = 0;
    int i;
    for(i = 1; i < MAX_LINE; i++)
    {
        if(s_lines[i].used == 0)
        {   
            fd = i;
            //初始化port
            bzero(&s_lines[i], sizeof(Line));
            s_lines[i].used = 1;
            break;
        }
    }
    js_sys("port", "new linefd:%d\n", fd);
    return fd;
}
int line_get_write_buffd(int fd){
	Line *line = fd2line(fd);
	return line->buf_write;
}
int line_is_close(int fd){
	Line *line = fd2line(fd);
	return line->used == 0;
}
static int line_free(int fd)
{
    Line *line = fd2line(fd);
    line->used = 0;
}

static int line_info(int linefd)
{
    Line *line = fd2line(linefd);
    int sockfd = line->sockfd;
    cbuf_info(line->buf_read);
    cbuf_info(line->buf_write);
}
static void port_on_accept(int listenfd, short event, void *args)
{
    Port* port = (Port *)args;	
	int sockfd;	
    int portfd;
	struct sockaddr_in addr;	
	socklen_t addrlen = sizeof(addr);	
    portfd = port->portfd;
    
    sockfd = accept(listenfd, (struct sockaddr*)&addr, &addrlen);
    js_sys("port", "a client connected ! portfd:%d sockfd:%d handler:%x\n", portfd, sockfd, port->cb_accept);
    if(port->cb_accept)
    {
        port->cb_accept(portfd, sockfd);
    }else
    {
        port_reject(portfd, sockfd);
    }
}
static int port_read_clear(linefd)
{
    Line *line = fd2line(linefd);
    int sockfd = line->sockfd;
    char buf[1024];
    int ir = read(sockfd, buf, 1024);
}
static void port_dispatch(int linefd)
{
    int i;
    Line *line = fd2line(linefd);
    int buffd = line->buf_read;
    int portfd = line->portfd;
    js_sys("port", "dispatch linefd:%d\n", linefd);
    for(;;)
    {
        
    	js_sys("port", "dispatch buffd:%d\n", buffd);

        int plen = cbuf_look_int(buffd);
        js_sys("port", "plen :%d\n", plen);
        int buf_size = cbuf_size(buffd);
        js_sys("port", "cbuf_size :%d\n", buf_size);
        if(plen > 10000){
            char *rbuf = cbuf_base(buffd);
            for(i = 0; i < 50; i++){
                printf("%d\n", rbuf[i]);
            }
            sleep(1);
        	int a = 1/0;
        }
        if(plen <= 0 || buf_size < plen)
        {
            js_sys("port", "port_dispatch is not finished\n");
            break;
        }

        
        Port *port = fd2port(portfd);

        char *rptr = cbuf_rptr(buffd);

        if(port->cb_packet)
        {
            int pktfd = pkt_new(buffd);
            int msgid = pkt2msgid(pktfd);
            port->cb_packet(portfd, linefd, msgid, pktfd);
            pkt_free(pktfd);
        }else if(port->lua_cb_packet[0] != 0){

        	int pktfd = pkt_new(buffd);
        	int msgid = pkt2msgid(pktfd);
        	js_sys("port", "lua callback %s\n", port->lua_cb_packet);
        	luas_classfunc(port->lua_cb_self, port->lua_cb_packet, "(%d, %d)",
        			msgid, pktfd);
        	pkt_free(pktfd); 
        }
        int size_read = buf_size - cbuf_size(buffd);
        //没读完的跳过
        cbuf_set_rptr(buffd, rptr);
        cbuf_rskip(buffd, plen);

        js_sys("port", "eat a packet\n");
    }
}
static void port_on_read(int sockfd, short event, void *args)
{

    int linefd;
    int i, j;
    linefd = (int)args;
    Line *line = fd2line(linefd);
    int buffd = line->buf_read;
    js_sys("port", "linefd:%d sockfd:%d\n", linefd, sockfd);

    //保证buf还剩下这么多
    if(cbuf_remain(buffd) < 1024)
    {
        cbuf_double(buffd);
    }
    //一次至少能读到多少,这个数值关系到系统调用read的次数
    if(cbuf_wlen(buffd) < 64)
    {
        cbuf_2line(buffd);
    }
    cbuf_info(buffd);
    int   wlen = cbuf_wlen(buffd);
    char* wbuf = cbuf_wptr(buffd);
    js_sys("port", "goto read %d byte\n", wlen);
    int rt = recv(sockfd, wbuf, wlen, 0);
//    for(j = 0; j < rt; j++){
//    	printf("%d ", wbuf[j]);
//    	if(wbuf[j] == 27){
//    	    printf("\n");
//    	}
//    }
//    printf("\n");
    js_sys("port", "system call read rt is:%d\n", rt);
    if(rt > 0)
    {
        cbuf_wskip(buffd, rt);
    }
    if(rt > 0 && rt < wlen)//读完了
    {       
        js_sys("port", "port_on_read read finished\n");
        goto dispatch;
    }
    if(rt < 0 && errno == EAGAIN)//读完了
    {
        js_sys("port", "port_on_read read finished\n");
        goto dispatch;
    }
    if(rt > 0 && rt == wlen)//里面还有,继续读
    {
        js_sys("port", "port_on_read continue\n");
        goto dispatch;
        //port_on_read(sockfd, event, args);
       // return;
    }
    if(rt == 0)
    {
        //close
        js_sys("port", "port_on_read close\n");
        close(sockfd);
        goto close;;
    }else //rt < 0
    {
        //error
        js_sys("port", "port_on_read error :%s\n", strerror(errno));
        goto close;;
    }    
dispatch:
    //port_read_clear(linefd);
    line_info(linefd);    
    js_sys("port", "dispatch to linefd:%d sockfd:%d\n", linefd, sockfd);
    port_dispatch(linefd);
    return;
close:
    port_close_line(linefd);
    return;
}
static void port_on_write(int sockfd, short event, void *args)
{
    int i, j;
    int linefd = (int)args;
    Line *line = fd2line(linefd);
    int buffd = line->buf_write;
    //分两段写
    for(i = 0; i < 2; i++)
    {
        int   rlen = cbuf_rlen(buffd);
        char* rbuf = cbuf_rptr(buffd);
        if(rlen > 0)
        {
            int ir = send(sockfd, rbuf, rlen, 0);

            //写满了?
            if(ir == -1 && errno == EAGAIN)
            {
                break;
            }
            //写满了
            if(ir == 0)
            {
                break;
            }

            if(ir > 0)
            {
                cbuf_rskip(buffd, ir);

            }
        }
    }
    //js_sys("port", "port_on_write linefd:%d, sockfd:%d\n", linefd, sockfd);
}

int port_send(int linefd, int msgid, const char* buf, int buf_len)
{
    if(linefd == 0){
        return 0;
    }
    Line *line = fd2line(linefd);
    int buffd = line->buf_write;
    cbuf_write_int(buffd, buf_len + 8);
    cbuf_write_int(buffd, msgid);
    cbuf_write(buffd, buf, buf_len);

    cbuf_info(buffd);
    return 1; 
}
int port_close_line(int linefd)
{

    Line *line = fd2line(linefd);
    int portfd = line->portfd;
    Port *port = fd2port(portfd);

    if(port->cb_close)
    {
    	port->cb_close(portfd, linefd);
    }
    int sockfd = line->sockfd;
    
    //移动事件
    event_del(&line->ev_write);
    event_del(&line->ev_read);
    //释放buff
    cbuf_free(line->buf_read);
    cbuf_free(line->buf_write);
    //关闭socket
    close(sockfd);
    //释放line
    line_free(linefd);

    js_sys("port", "close line portfd:%d, linefd:%d, sockfd:%d\n", portfd, linefd, sockfd);
}
static void setnonblock(int sockfd)
{
  int flags;

  flags = fcntl(sockfd, F_GETFL);
  flags |= O_NONBLOCK;
  fcntl(sockfd, F_SETFL, flags);
}
int port_get_by_name(const char *name)
{
    void *rt = dictFetchValue(s_name_port, (void*)name);

    //todo
    return (int)rt;
}
int port_rename(int fd, const char *name)
{
    return dictReplace(s_name_port, (void*)name, (void *)fd);

}
int port_accept(int portfd, int sockfd)
{
    int linefd = line_new();
    if(linefd <= 0)
    {   
        js_sys("port", "port_accept line is full portfd:%d, sockfd:%d\n", portfd, sockfd);
        close(sockfd); 
        return 0;
    }
    Line *line = fd2line(linefd);
    
    setnonblock(sockfd);
    //初始化
    line->sockfd = sockfd;
    line->portfd = portfd;
    line->buf_read = cbuf_new(1024);
    line->buf_write = cbuf_new(1024);
    
    event_set(&line->ev_read,
                line->sockfd,
                EV_READ | EV_PERSIST,
                port_on_read,
                (void *)linefd);
    event_set(&line->ev_write,
                line->sockfd,
                EV_WRITE | EV_PERSIST,
                port_on_write,
                (void *)linefd);
                
    event_add(&line->ev_write, NULL);
    event_add(&line->ev_read, NULL);
    return linefd;
}
int port_reject(int portfd, int sockfd)
{
    js_sys("port", "reject portfd:%d, sockfd:%d\n", portfd, sockfd);
    close(sockfd); 
}
int port_new()
{
    int fd = 0;
    int i;
    for(i = 1; i < MAX_PORT; i++)
    {
        if(s_ports[i].used == 0)
        {   
            fd = i;
            //初始化port
            bzero(&s_ports[i], sizeof(Port));
            s_ports[i].used = 1;
            s_ports[i].portfd = fd;
            break;
        }
    }
    js_sys("port", "new portfd:%d\n", fd);
    return fd;
}
static int port_free(int portfd){
	Port *port = fd2port(portfd);
	port->used = 0;
}
void  port_set_userdata(int portfd, void *userdata)
{
    Port *port = fd2port(portfd);
    port->userdata = userdata;
}
void *port_get_userdata(int portfd)
{
    Port *port = fd2port(portfd);
    return port->userdata;
}
int port_set_cb_accept(int portfd, port_cb_accept cb)
{
    Port *port = fd2port(portfd);
    port->cb_accept = cb;
    return 1;
}
int port_set_lua_cb_packet(int portfd, int self, const char *funcname){
	js_sys("port", "port_set_lua_cb_packet %s\n", funcname);
	Port *port = fd2port(portfd);
	strcpy(port->lua_cb_packet, funcname);
	port->lua_cb_self = self;
	return 1;
}
int port_set_cb_close(int portfd, port_cb_close cb)
{
    Port *port = fd2port(portfd);
    port->cb_close = cb;
    return 1;
}
int port_set_cb_packet(int portfd, port_cb_packet cb)
{
    Port *port = fd2port(portfd);
    port->cb_packet = cb;
    return 1;
}
int port_connect_addr(int portfd, struct sockaddr_in addr)
{
    unsigned short portno = ntohs(addr.sin_port);
    char *ip = inet_ntoa(addr.sin_addr);

    js_sys("port", "port_connect %s:%d\n", ip, portno);
    
    if(portfd <= 0)
    {
        js_sys("port", "port_connect portfd invalid\n");
        return 0;
    }
    
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0)
    {
        js_sys("port", "port_connect create socket fail\n");
        return 0;
    }

    int rt = connect(sockfd, (struct sockaddr *)&addr, sizeof(addr));
    if(rt < 0)
    {
        js_sys("port", "port_connect fail errno:%d:%s\n", errno, strerror(errno));
        return 0;
    }

    int linefd = line_new();
    if(linefd <= 0)
    {   
        js_sys("port", "port_connect line is full portfd:%d\n", portfd);
        return 0;
    }
    Line *line = fd2line(linefd);
    
    setnonblock(sockfd);
    //初始化
    line->sockfd = sockfd;
    line->portfd = portfd;
    line->buf_read = cbuf_new(1024);
    line->buf_write = cbuf_new(1024);
    
    event_set(&line->ev_read,
                line->sockfd,
                EV_READ | EV_PERSIST,
                port_on_read,
                (void *)linefd);
    event_set(&line->ev_write,
                line->sockfd,
                EV_WRITE | EV_PERSIST,
                port_on_write,
                (void *)linefd);
                
    event_add(&line->ev_write, NULL);
    event_add(&line->ev_read, NULL);
    return linefd;
}
int port_connect(int portfd, const char *ip, unsigned short portno)
{
    struct sockaddr_in server_addr;
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = inet_addr(ip);   
    server_addr.sin_port = htons(portno);
    return port_connect_addr(portfd, server_addr);
}
int port_select(int portfd, int base)
{
    Port *port = fd2port(portfd);
    int linenum = port->load_line_num;
    int idx = base % linenum;
    LoadLine *loadline = &(port->load_lines[idx]);
    js_log("port", "linefd:%d\n", loadline->linefd);
    if(loadline->linefd <= 0)
    {
        loadline->linefd = port_connect_addr(portfd, loadline->addr);
    }
    return loadline->linefd;
}
int port_listen(int portfd, unsigned short portno)
{
    int listenfd;
    int reuse = 1;
	struct sockaddr_in addr;
    Port *port;
    
    port = fd2port(portfd);
	listenfd = socket(AF_INET, SOCK_STREAM, 0);	
	if(listenfd < 0)	
	{		
		js_sys("port", "failed to socket on port :%d\n", portno);
		port_free(portfd);
		return 0;	
	}	
	bzero((void*)&addr, sizeof(addr));	
	addr.sin_family = AF_INET;	
	addr.sin_addr.s_addr = INADDR_ANY;	
	addr.sin_port = htons(portno);	
	
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
    
	if(bind(listenfd,(struct sockaddr *)&addr,sizeof(addr)) < 0)	
	{		
		js_sys("port", "fail to bind on port:%d\n", portno);
		port_free(portfd);
		return 0;	
	}	
	if(listen(listenfd, 5) < 0)
	{		
		js_sys("port", "fail to listen on port:%d\n", portno);
		port_free(portfd);
		return 0;	
	}
    
    port->listenfd = listenfd;
     
	event_set(&(port->evt), 
			listenfd, 
			EV_READ | EV_PERSIST,
			port_on_accept,
			port);
	event_add(&(port->evt), NULL);       
    js_sys("port", "listen on port:%d\n", portno);
    return 1;
}





static int Listen(lua_State *L)
{
    if (LUA_ARGNUM == 2 && LUA_ISNUMBER(1) && LUA_ISNUMBER(2))
    {
        int portfd = (int)LUA_TONUMBER(1);
        unsigned short portno = (unsigned short)LUA_TONUMBER(2);
        int rt = port_listen(portfd, portno);
        LUA_RETURN(rt);
    }else{
    	js_sys("port", "interface Listen arg invalid\n");
    }

    LUA_RETURN(0);
}
static int Connect(lua_State *L)
{
    if (LUA_ARGNUM == 3 && LUA_ISNUMBER(1) && LUA_ISSTRING(2) && LUA_ISNUMBER(3))
    {
        int portfd = (int)LUA_TONUMBER(1);
        const char *ip = (const char *)LUA_TOSTRING(2);
        unsigned short portno = (unsigned short)LUA_TONUMBER(3);
        int rt = port_connect(portfd, ip, portno);
        LUA_RETURN(rt);
    }
 
    LUA_RETURN(0);
}
static int ConnectV(lua_State *L)
{   
    int i ;
    int argnum = LUA_ARGNUM;
    int portfd = (int)LUA_TONUMBER(1);
    Port *port = fd2port(portfd);
    //连接池
    int linenum = (argnum-1) / 2;
    int mem_size = sizeof(LoadLine) * linenum;
    port->load_lines = (LoadLine *)malloc(mem_size);
    port->load_line_num = linenum;
    for(i = 0; i < linenum; i++)
    {
        LoadLine *loadline = &(port->load_lines[i]);
        const char *ip = (const char *)LUA_TOSTRING(i * 2 + 2);
        unsigned short portno = (unsigned short)LUA_TONUMBER( i * 2 + 3);

        loadline->addr.sin_family = AF_INET;
        loadline->addr.sin_addr.s_addr = inet_addr(ip);   
        loadline->addr.sin_port = htons(portno);        
        loadline->linefd = port_connect(portfd, ip, portno);
        
        LUA_RETURN(1);
    }
 
    LUA_RETURN(0);
}
static int Select(lua_State *L)
{
    if (LUA_ARGNUM == 2 && LUA_ISNUMBER(1) && LUA_ISNUMBER(2) )
    {
        int portfd = (int)LUA_TONUMBER(1);
        int base = (int)LUA_TONUMBER(2);
        int rt = port_select(portfd, base);
        LUA_RETURN(rt);
    }
 
    LUA_RETURN(0);
}
static int Rename(lua_State *L)
{
    if (LUA_ARGNUM == 2 && LUA_ISNUMBER(1) && LUA_ISSTRING(2))
    {
        int portfd = (int)LUA_TONUMBER(1);
        const char * name = (const char *)LUA_TOSTRING(2);
        int rt = port_rename(portfd, name);
        LUA_RETURN(rt);
    }

    LUA_RETURN(0);
}
static int GetByName(lua_State *L)
{
    if (LUA_ARGNUM == 1 && LUA_ISSTRING(1))
    {
        const char * name = (const char *)LUA_TOSTRING(1);
        int rt = port_get_by_name(name);
        LUA_RETURN(rt);
    }

    LUA_RETURN(0);
}
static int OnPacket(lua_State *L)
{
	js_sys("port", "OnPacket\n");
    if (LUA_ARGNUM == 3 && LUA_ISNUMBER(1) && LUA_ISSTRING(2))
    {
        int fd 					= (int)LUA_TONUMBER(1);
        const char *funcname 	= (const char *)LUA_TOSTRING(2);
        int self                = (int)luaL_ref(L, LUA_REGISTRYINDEX);
        int rt = port_set_lua_cb_packet(fd, self, funcname);
        LUA_RETURN(rt);
    }

    LUA_RETURN(0);
}
static int Send(lua_State *L)
{
    int i;
	int arg_num = LUA_ARGNUM;
    
    
	if (arg_num >= 2 && LUA_ISNUMBER(1) && LUA_ISNUMBER(2) )
	{

        int linefd = (int)LUA_TONUMBER(1);

        int buffd = line_get_write_buffd(linefd);
        int body_len = 0;
        unsigned int msgid = (unsigned int) LUA_TONUMBER(2);
        char *wptr_len = cbuf_wptr(buffd);
        cbuf_write_int(buffd, 0);
        cbuf_write_int(buffd, msgid);

        for (i = 3; i <= arg_num; i++)
        {
            if (LUA_ISSTRING(i))
            {
            	char *str = (char *)LUA_TOSTRING(i);
                cbuf_write_utf(buffd, str);
                body_len += strlen(str) + 2;
            }else if (LUA_ISNUMBER(i))
            {
                cbuf_write_int(buffd, (int) LUA_TONUMBER(i));
                body_len += sizeof(int);
            }
        }
        char *wptr_now = cbuf_wptr(buffd);
        cbuf_set_wptr(buffd, wptr_len);
        cbuf_write_int(buffd, body_len + 8);
        cbuf_set_wptr(buffd, wptr_now);

        cbuf_info(buffd);
        js_sys("port", "send %d bytes\n", body_len);


	}

    LUA_RETURN(1);
}

static int New(lua_State *L)
{
    if (LUA_ARGNUM == 0)
    {
        int portfd = port_new();
        LUA_RETURN(portfd);
    }
    LUA_RETURN(-1);
}

static luaL_Reg lua_lib[] = 
{
    LUA_FUNC_INIT(Listen),
    LUA_FUNC_INIT(Connect),
    LUA_FUNC_INIT(ConnectV),
    LUA_FUNC_INIT(Select),
    LUA_FUNC_INIT(Rename),
    LUA_FUNC_INIT(GetByName),
    LUA_FUNC_INIT(Send),
    LUA_FUNC_INIT(OnPacket),
    LUA_FUNC_INIT(New),
    {NULL, NULL}
};

int port_init()
{
    s_name_port = dictCreate(&dictTypeHeapStrings, "name_port");
    luas_reg_lib("Port", lua_lib);

}
